package com.bd03.cmccmonitor.dbSql

import com.bd03.cmccmonitor.conf.{ConfigHelper, Constant}
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import scalikejdbc._

object OffsetHandler {
  //从数据库读取偏移量
  def obtainOffset4Mysql={
    DB.readOnly{ implicit sesson=>
      sql"select * from cmcc_offset where groupID = ? and topics = ?"
        .bind(Constant.GROUPID,Constant.TOPIC)
        .map(t=>(new TopicPartition( t.string("topics"),t.int("partition")), t.long("offsets")))
        .list().apply()
    }.toMap
  }
//将新的偏移量写入到数据库
  def savaOffset2Mysql(ranges:Array[OffsetRange])={
    DB.localTx { implicit session =>
      ranges.foreach(t => {
        sql"replace into cmcc_offset values(?,?,?,?)"
          .bind(ConfigHelper.GROUPID,t.topic,t.partition,t.untilOffset)
          .update()
          .apply()
      })
    }
  }

}
